package cn.azzhu.day02

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @author azzhu
 * @create 2020-09-20 23:57:32
 */
object FlatMapImplementsMapAndFilter {
  def main(args: Array[String]): Unit = {
    //todo 1.获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //todo 2.数据源
    val stream = env.addSource(new SensorSource)

    // 使用flatMap实现map功能，将传感器数据的id抽取出来
    stream.flatMap(new FlatMapFunction[SensorReading,String] {
      override def flatMap(value: SensorReading, out: Collector[String]): Unit = out.collect(value.id)
    }).print()

    //使用flatMap实现filter的功能,将sensor_1的数据过滤出来
    stream.flatMap(new FlatMapFunction[SensorReading,SensorReading] {
      override def flatMap(value: SensorReading, out: Collector[SensorReading]): Unit = {
        if(value.id.equals("sensor_1")) {
          out.collect(value)
        }
      }
    })

    env.execute("FlatMapExample2")
  }


}
